[AWS re:Invent] Workshop : การเร่งพัฒนา API แบบ Serverless ด้วย AWS Lambda Powertools

ผมได้เข้าร่วม Session ในรูปแบบ workshop ที่จัดขึ้นในงาน AWS re:Invent 2024 หัวข้อ "SVS306: Accelerate development with AWS Lambda Powertools for serverless APIs" จึงอยากจะมาแชร์ประสบการณ์นี้

#AWS

#Classmethod Thailand

#Thai Language

#AWS Lambda

#Lambda Powertools

Patarapong Armuttaporn

2024.12.27

บทความนี้แปลมาจากบทความภาษาญี่ปุ่นที่ชื่อว่า [レポート]AWS Lambda Powertoolsを活用してサーバーレスAPIの開発を加速するワークショップに参加しました #SVS306 #AWSreInvent โดยเจ้าของบทความนี้คือ คุณ 塚本太朗

บทนำ

สวัสดีครับ ผมสึคาโมโตะ จากแผนก Retail App Co-Creation

ผมได้เข้าร่วม workshop ในงาน AWS re:Invent 2024 ในหัวข้อ "SVS306: Accelerate development with AWS Lambda Powertools for serverless APIs" จึงอยากจะมาแชร์ประสบการณ์นี้

แม้ว่าจะเป็นครั้งแรกที่ผมเข้าร่วมกิจกรรมในรูปแบบ workshop และยังไม่คุ้นเคยกับขั้นตอนต่างๆ แต่ผมก็สามารถทำตามขั้นตอนได้เกือบทั้งหมด
นอกจากนี้ การได้ถามคำถามเป็นภาษาอังกฤษก็เป็นประสบการณ์ที่ดี
workshop ประกอบด้วย 3 ขั้นตอน แต่ผมทำได้เพียงถึงขั้นตอนที่ 2 เท่านั้น

Session Information

Session ID: SVS306
Title: Accelerate development with AWS Lambda Powertools for serverless APIs
Level: 300 - Advanced
ระยะเวลา: 120 นาที
กลุ่มเป้าหมาย:
- DevOps Engineer
- Developer / Engineer
- IT Professional
- Technical Manager

Session Overview

[คำอธิบาย Session อย่างเป็นทางการ]

In this workshop, start with an existing application built with Python and progressively improve your API event handler using Powertools for AWS Lambda. Learn how to implement request and response validation, dynamic routing, exception handling, middleware, and OpenAPI schema generation. Discover how to improve your API event handler with serverless best practices using Python that you can easily extend to other Powertools runtimes. You must bring your laptop to participate.

[แปล]

ใน workshop นี้ เริ่มต้นด้วยแอปพลิเคชันที่มีอยู่แล้วซึ่งสร้างด้วย Python และพัฒนา API event handler ของคุณอย่างต่อเนื่องโดยใช้ Powertools สำหรับ AWS Lambda คุณจะได้เรียนรู้วิธีการติดตั้งการตรวจสอบความถูกต้องของ request และ response, dynamic routing, exception handling, middleware และการสร้าง OpenAPI schema นอกจากนี้ คุณจะได้ค้นพบวิธีการปรับปรุง API event handler ด้วย serverless best practices โดยใช้ Python ซึ่งคุณสามารถขยายไปยัง Powertools runtimes อื่นๆ ได้อย่างง่ายดาย ผู้เข้าร่วมจำเป็นต้องนำแล็ปท็อปมาด้วย

จุดเด่น

เป็น workshop ที่แสดงการปรับปรุงโครงสร้าง serverless ทั่วไปทีละขั้นตอนโดยใช้ AWS Lambda Powertools
คุณจะได้เรียนรู้กระบวนการปรับปรุงในด้านการจัดการข้อผิดพลาดและ observability
เป็น Session ที่แนะนำสำหรับผู้ที่เพิ่งเริ่มสร้างโครงสร้าง serverless ด้วย Python หรือผู้ที่ต้องการเรียนรู้เชิงลึกเกี่ยวกับ AWS Lambda Powertools
เนื่องจากเป็น workshop ที่เน้นการเขียนโค้ดเป็นหลัก จึงอาจไม่เหมาะสำหรับผู้ที่ต้องการทำ workshop ที่ใช้ AWS Management Console

โครงสร้างของ Workshop

Create your first API
- What is a Lambda integration
- Understanding the legacy application
- Adding Powertools for AWS Lambda (Python)
- Defining routes with Powertools 4.1. Using the Router Object for better organization
- Using different HTTP methods
- Using the Response object for consistent API responses
- Adding CORS protection
- Handling exceptions and not found routes
- Observing your API
- Securing your API
Middleware, Data Validation & Idempotency
- Using Middleware
- Using Data validation
- Ensuring idempotency
OpenApi

ภาพรวมของ "Create your first API"

เริ่มต้นจากการ implement ที่ไม่ได้ใช้ AWS Lambda Powertools และค่อยๆ ปรับปรุงให้ดีขึ้น
ตัวอย่างเช่น มีการใช้ฟีเจอร์ต่างๆ ของ Lambda Powertools ดังนี้

Router: กำหนดเส้นทางคำขอจาก APIGateway ตาม path และ method ตัวอย่างการใช้งานมีดังนี้

@router.delete("/orders/<order_id>")
def delete_order(order_id: str):
    orders_table.delete_item(Key={'orderId': order_id})

    return Response(
        status_code=HTTPStatus.NO_CONTENT.value,  # HTTP CODE 204
        content_type=content_types.APPLICATION_JSON,
    )

Response: สามารถสร้าง response ในรูปแบบที่จะส่งคืนให้ APIGateway ได้อย่างง่ายดาย

return Response(
    status_code=HTTPStatus.OK.value,  # HTTP CODE 200
    content_type=content_types.APPLICATION_JSON,
    body=response['Item'],
)

CORSConfig: สามารถกำหนดค่าการตั้งค่าที่เกี่ยวข้องกับ CORS (Cross-Origin Resource Sharing) ได้

cors_config = CORSConfig(allow_origin="https://www.amazon.com", max_age=300)
app = APIGatewayRestResolver(cors=cors_config)
app.include_router(router)

metric: สามารถตรวจสอบสถิติต่างๆ ผ่าน CloudWatch metrics ได้

@metrics.log_metrics(capture_cold_start=True)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

tracer: เพิ่มข้อมูลรายละเอียดในการติดตามของ X-Ray

tracer = Tracer()
@tracer.capture_lambda_handler
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

ภาพรวมของ "Middleware, Data Validation & Idempotency"

ใช้ Middleware เพื่อจำกัดผู้ใช้ที่สามารถเรียกใช้ API ได้
โดยการใช้ Middleware เป็น decorator จะสามารถทำให้มีการประมวลผลก่อนการเรียกใช้ API ทุกครั้ง
ใน workshop ครั้งนี้ เราได้ใช้ Middleware เป็น decorator เพื่อสร้างกรณีที่อนุญาตให้เฉพาะผู้ใช้บางรายเท่านั้นที่สามารถเรียกใช้ API ได้

การเขียนโค้ดในส่วนที่กำหนด middleware

app = APIGatewayRestResolver(cors=cors_config)
app.include_router(router)
app.use(middlewares=[restrict_data_access]) # ⭐️ การตั้งค่า middleware

การเขียนโค้ดของตัว middleware เอง

def restrict_data_access(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
    if "authorization" in app.current_event.headers:
        # Decode the token
        token = app.current_event.headers["authorization"]
        decoded_token = json.loads(base64.b64decode(token).decode('utf-8'))
        #===ละไว้===#

    response = next_middleware(app)
    return response

ความรู้สึกจากการเข้าร่วม workshop ครั้งแรก

เนื่องจากปกติผมใช้ AWS Lambda Powertools กับ TypeScript อยู่แล้ว จึงไม่มีฟีเจอร์ใหม่ๆ แต่ก็ได้ทบทวนความรู้
นอกจากนี้ แม้จะรู้สึกประหม่าเล็กน้อยเพราะเป็นการเข้าร่วม workshop ครั้งแรก แต่สามารถดำเนินการผ่านเว็บไซต์ของ workshop ได้ง่ายกว่าที่คิด
โอกาสที่จะได้ทำ workshop ในสถานที่จริงโดยไม่ต้องกังวลเรื่องค่าใช้จ่ายนั้นมีน้อย ดังนั้นผมจึงอยากเข้าร่วม workshop ให้มากขึ้นในอนาคต

Final implementation (สำหรับอ้างอิง)

แม้จะเป็นเพียงส่วนที่ทำได้จนจบ Session แต่การ implement มีลักษณะดังต่อไปนี้

app.py: Lambda entry point

#####
# imports - app.py
#####
from http import HTTPStatus
import json
import base64
from aws_lambda_powertools.event_handler import APIGatewayRestResolver, CORSConfig
from aws_lambda_powertools.event_handler import (
    Response,
    content_types,
)
from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.event_handler.exceptions import NotFoundError
from aws_lambda_powertools import Logger, Metrics, Tracer
from all_routes import router
from middleware import restrict_data_access

#####
# Classes, functions and instances - app.py
#####
logger = Logger()
metrics = Metrics()
tracer = Tracer()

cors_config = CORSConfig(allow_origin="https://www.amazon.com", max_age=300)
app = APIGatewayRestResolver(cors=cors_config)
app.include_router(router)
app.use(middlewares=[restrict_data_access])

@app.exception_handler([ValueError, AttributeError])
def handle_invalid_payload(ex: ValueError | AttributeError):
    metadata = {"path": app.current_event.path, "http_method": app.current_event.http_method}
    logger.exception(f"Malformed request: {ex}", metadata=metadata)

    return Response(
        status_code=HTTPStatus.BAD_REQUEST.value,
        content_type=content_types.APPLICATION_JSON,
        body={"message": "Invalid request parameters. Please verify your parameters or payload according to our documentation."}
    )

@app.not_found
def handle_not_found_errors(exc: NotFoundError) -> Response:
    logger.info(f"Route not found: {app.current_event.path}")
    return Response(status_code=HTTPStatus.NOT_FOUND.value, content_type=content_types.TEXT_PLAIN, body="Sorry, I don't exist!")

#####
# Lambda handler - app.py
#####
@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics(capture_cold_start=True)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

all_routes.py

#####
# imports - all_routes.py
#####
from http import HTTPStatus
from uuid import uuid4
import boto3
import json
from aws_lambda_powertools import Logger, Metrics, Tracer
from aws_lambda_powertools.metrics import MetricUnit
from aws_lambda_powertools.event_handler.api_gateway import Router
from aws_lambda_powertools.event_handler import (
    Response,
    content_types,
)

#####
# Classes, functions and instances - all_routes.py
#####
logger = Logger()
metrics = Metrics()
tracer = Tracer()

dynamodb = boto3.resource('dynamodb')
orders_table = dynamodb.Table('OrdersWorkshop')
router = Router()

#####
# Get all orders method - all_routes.py
#####
@router.get("/orders")
def get_all_orders():
    response = orders_table.scan()

    if len(response['Items']) > 0:
        return Response(
            status_code=HTTPStatus.OK.value,  # HTTP CODE 200
            content_type=content_types.APPLICATION_JSON,
            body=response['Items'],
        )
    else:
        return Response(
            status_code=HTTPStatus.NOT_FOUND.value,  # HTTP CODE 404
            content_type=content_types.APPLICATION_JSON,
            body={"message": "No orders found"}
        )

#####
# Get order method - all_routes.py
#####
@router.get("/orders/<order_id>")
@tracer.capture_method
def get_order(order_id: str):
    response = orders_table.get_item(Key={'orderId': order_id})

    # Logging
    logger.info("Searching an order", order_id=order_id)

    # Adding metric
    metrics.add_dimension(name="order_id", value=order_id)
    metrics.add_metric("OrderSearch", unit=MetricUnit.Count, value=1)

    if 'Item' in response:
        return Response(
            status_code=HTTPStatus.OK.value,  # HTTP CODE 200
            content_type=content_types.APPLICATION_JSON,
            body=response['Item'],
        )
    else:
        return Response(
            status_code=HTTPStatus.NOT_FOUND.value,  # HTTP CODE 404
            content_type=content_types.APPLICATION_JSON,
            body={"message": "Order not found"},
        )

#####
# Create order method - all_routes.py
#####
@router.post("/orders")
def create_order():
    body = router.current_event.json_body
    order_id = str(uuid4())

    item = {
        'orderId': order_id,
        'customerName': body.get('customerName'),
        "restaurantName": body.get('restaurantName'),
        'orderItems': body.get('orderItems'),
        'orderDate': body.get('orderDate'),
        'orderStatus': 'Pending',
        'restaurantId': body.get('restaurantId'),
    }

    orders_table.put_item(Item=item)

    return Response(
        status_code=HTTPStatus.CREATED.value,  # HTTP CODE 201
        content_type=content_types.APPLICATION_JSON,
        headers={"Location": f"/orders/{order_id}"},
        body=item,
    )

#####
# Update order method - all_routes.py
#####
@router.put("/orders/<order_id>")
def update_order(order_id: str):
    body = router.current_event.json_body

    response = orders_table.update_item(
        Key={'orderId': order_id},
        UpdateExpression='SET customerName = :name, orderItems = :items, orderDate = :date, orderStatus = :status',
        ExpressionAttributeValues={
            ':name': body.get('customerName'),
            ':items': body.get('orderItems'),
            ':date': body.get('orderDate'),
            ':status': body.get('orderStatus'),
        },
        ReturnValues='ALL_NEW'
    )

    return Response(
        status_code=HTTPStatus.OK.value,  # HTTP CODE 200
        content_type=content_types.APPLICATION_JSON,
        body=response['Attributes'],
    )

#####
# Delete order method - all_routes.py
#####
@router.delete("/orders/<order_id>")
def delete_order(order_id: str):
    orders_table.delete_item(Key={'orderId': order_id})

    return Response(
        status_code=HTTPStatus.NO_CONTENT.value,  # HTTP CODE 204
        content_type=content_types.APPLICATION_JSON,
    )

#####
# Get all orders per restaurant method - all_routes.py
#####
@router.get("/orders_per_restaurant/<restaurant_id>")
def get_all_orders_per_restaurant(restaurant_id: str):

    response = orders_table.scan(
        FilterExpression='restaurantId = :rid',
        ExpressionAttributeValues={
            ':rid': restaurant_id
        }
    )

    if len(response['Items']) > 0:
        return Response(
            status_code=HTTPStatus.OK.value,  # HTTP CODE 200
            content_type=content_types.APPLICATION_JSON,
            body=response['Items'],
        )
    else:
        return Response(
            status_code=HTTPStatus.NOT_FOUND.value,  # HTTP CODE 404
            content_type=content_types.APPLICATION_JSON,
            body={"message": "No orders found"},
        )

middleware.py

#####
# imports - middleware.py
#####
from http import HTTPStatus
import json
import base64
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.event_handler import (
    Response,
    content_types,
)
from aws_lambda_powertools.event_handler.middlewares import NextMiddleware
from aws_lambda_powertools.utilities.typing import LambdaContext

#####
# Middleware function - middleware.py
#####
def restrict_data_access(app: APIGatewayRestResolver, next_middleware: NextMiddleware) -> Response:
    if "authorization" in app.current_event.headers:
        # Decode the token
        token = app.current_event.headers["authorization"]
        decoded_token = json.loads(base64.b64decode(token).decode('utf-8'))

        # Admin should return early, full privileges
        if decoded_token.get("level") == "admin":
            return next_middleware(app)

        restaurant_id = str(decoded_token.get("restaurant_id"))
        path_params = app.context.get("_route_args", {})

        if "restaurant_id" in path_params:
            if str(path_params["restaurant_id"]) != restaurant_id:
                return Response(
                    status_code=HTTPStatus.FORBIDDEN.value,
                    content_type=content_types.APPLICATION_JSON,
                    body={"message": f"Access denied: You don't have permission for restaurant {path_params['restaurant_id']}"}
                )
            return next_middleware(app)

        response = next_middleware(app)

        if str(response.body.get("restaurantId")) != restaurant_id:
            return Response(
                status_code=HTTPStatus.FORBIDDEN.value,
                content_type=content_types.APPLICATION_JSON,
                body={"message": "Access denied: You don't have permission to access this order"}
            )

    response = next_middleware(app)
    return response